package com.whq.thrift.api.kafka.thriftlog;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.ibatis.session.ExecutorType;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.defaults.DefaultSqlSessionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.alibaba.fastjson.JSON;
import com.whq.common.StringUtil;
import com.whq.thrift.api.common.constants.CommonConfigConstants;
import com.whq.thrift.api.common.spring.ReloadableConfManager;
import com.whq.thrift.api.kafka.thriftlog.db.ConsumeLog;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;

public class MainKafkaConsumerLog {
	private static final Logger LOGGER = LoggerFactory.getLogger(MainKafkaConsumerLog.class);
	private static final Logger LOGGER_QUARTZ = LoggerFactory.getLogger(CommonConfigConstants.LOGGER_KEY_QUARTZ);

	private final ConsumerConnector consumer;
	private ReloadableConfManager reloadableConfManager = null;
//	private KafkaService consumeLogService = null;
	// private IceClientHelper iceClientHelper = null;
	private String topic = "";
//	private ThreadPoolExecutor executor = null;
	private ClassPathXmlApplicationContext ctx = null;
	private DefaultSqlSessionFactory factory = null;

	public MainKafkaConsumerLog() {
		// 加载spring配置
		ctx = new ClassPathXmlApplicationContext("classpath:/conf/spring/applicationContext-core.xml");
		reloadableConfManager = (ReloadableConfManager) ctx.getBean("reloadableConfManager");
//		consumeLogService = ctx.getBean(KafkaService.class);
		factory = (DefaultSqlSessionFactory) ctx.getBean("sqlSessionFactory"); 
		// iceClientHelper = (IceClientHelper) ctx.getBean("iceClientHelper");
		topic = reloadableConfManager.getPropertyValue("kafka.consumer.receiveconsumelog.topic", null);

		Properties props = new Properties();
		// zookeeper 配置
		props.put("zookeeper.connect", reloadableConfManager.getPropertyValue("kafka.consumer.receiveconsumelog.zookeeper.connect", null));

		// group 代表一个消费组
		props.put("group.id", reloadableConfManager.getPropertyValue("kafka.consumer.receiveconsumelog.group.id", null));

		// zk连接超时
		props.put("zookeeper.session.timeout.ms", reloadableConfManager.getPropertyValue("kafka.consumer.receiveconsumelog.zookeeper.session.timeout.ms", null));
		props.put("zookeeper.sync.time.ms", reloadableConfManager.getPropertyValue("kafka.consumer.receiveconsumelog.zookeeper.sync.time.ms", null));
		props.put("auto.commit.interval.ms", reloadableConfManager.getPropertyValue("kafka.consumer.receiveconsumelog.auto.commit.interval.ms", null));
		props.put("auto.offset.reset", reloadableConfManager.getPropertyValue("kafka.consumer.receiveconsumelog.auto.offset.reset", null));
		// 序列化类
		props.put("serializer.class", "kafka.serializer.StringEncoder");

		ConsumerConfig config = new ConsumerConfig(props);

		consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);

//		// 初始化处理任务的线程池
//		// 保留的线程池大小
//		int corePoolSize = reloadableConfManager.getIntPropertyValue("saveloginlog.threadpool.concurrentThreadCount", null, 2);
//		// 线程池的最大大小
//		int maximumPoolSize = reloadableConfManager.getIntPropertyValue("saveloginlog.threadpool.threadQueueSize", null, 2);
//		// 空闲线程结束的超时时间
//		long keepAliveTime = reloadableConfManager.getIntPropertyValue("saveloginlog.threadpool.threadKeepAliveSeconds", null, 600);
//		// keepAliveTime 的单位
//		TimeUnit keepAliveTimeUnit = TimeUnit.SECONDS;
//
//		// 存放任务的队列
//		final BlockingQueue<Runnable> queueHandleTask = new LinkedBlockingQueue<Runnable>();
//		executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, keepAliveTimeUnit, queueHandleTask);
	}

	public void consume() {
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic, new Integer(1));

		StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
		StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());

		Map<String, List<KafkaStream<String, String>>> consumerMap = consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
		KafkaStream<String, String> stream = consumerMap.get(topic).get(0);

		ConsumerIterator<String, String> it = stream.iterator();
		final List<ConsumeLog> list = new ArrayList<ConsumeLog>(0);
		int countMs = reloadableConfManager.getIntPropertyValue("batch.save.record.maxcount", null, 100);
		long sleepMs = reloadableConfManager.getLongPropertyValue("sleep.ms.get.kafka.msg", null, 5);
		long intervalMs = reloadableConfManager.getLongPropertyValue("batch.save.max.interval.ms", null, 1000);
		long startMs = System.currentTimeMillis();

		Integer lastTableSeq = null;
		while (it.hasNext()) {
			String msg = it.next().message();
			if (StringUtil.isBlank(msg)) {
				continue;
			}
			ConsumeLog record = null;
			
			try {
				record = JSON.parseObject(msg, ConsumeLog.class);
			} catch (Exception e) {
				LOGGER.error("parse [" + msg + "] to ConsumeLog exception: ", e);
			}
			
			if (record == null) {
				continue;
			}
			
			record.setCreatetTime(new Date());
			
			int tableSeq = record.getTableSeq();
			
			list.add(record);
			
			long diffMs = System.currentTimeMillis() - startMs;
			boolean runInsertOrder = false;
			if ((list.size() >= countMs) || (diffMs >= intervalMs) || ((lastTableSeq != null) && (lastTableSeq != tableSeq))) {
				if (list.size() > 0) {
					//批量插入数据
					try {
						SqlSession sqlSession = factory.openSession(ExecutorType.BATCH, true);
						Map<String, Object> tmp = new HashMap<String, Object>();
						tmp.put("consumeLogList", list);
						tmp.put("tableSeq",	tableSeq);
					
						sqlSession.insert("ConsumeLogDao.insertBatch", tmp);
//						sqlSession.commit();
						sqlSession.close();
					} catch (Exception e) {
						LOGGER.error("insert records to db exception: ", e);
					}
					
					LOGGER_QUARTZ.info("insert record count: " + list.size());
				}

				list.clear();
				
				runInsertOrder = true;
				startMs = System.currentTimeMillis();
			}

			lastTableSeq = tableSeq;

			if (runInsertOrder) {
				try {
					Thread.sleep(sleepMs);
				} catch (Exception e) {
					LOGGER.error("Thread.sleep() exception: ", e);
				}
			}
		}

	}

	public void closeSpringCtx() {
		if (ctx != null) {
			ctx.close();
		}
	}

	public static void main(String[] args) {
		MainKafkaConsumerLog kafkaConsumerLog = new MainKafkaConsumerLog();
		kafkaConsumerLog.consume();
		kafkaConsumerLog.closeSpringCtx();
	}
}
